package com.qch.kafka.demo.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @Author qch
 * @Date 2021/5/26
 */
public class MyProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put("bootstrap.servers","192.168.142.135:9092");
        configs.put("key.serializer",IntegerSerializer.class);
        configs.put("value.serializer",StringSerializer.class);
       // configs.put("ack","1");
        //List<Header>headers=new ArrayList<Header>();
        //headers.add(new RecordHeader("biz.name","producer.demo".getBytes()));
        KafkaProducer<Integer, String> stringStringKafkaProducer = new KafkaProducer<Integer, String>(configs);
        ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(
                "topic_4",
                0,
                0,
                "hello"
        );
        //消息的同步确认
        Future<RecordMetadata> future = stringStringKafkaProducer.send(record);
        RecordMetadata recordMetadata = future.get();
        System.out.println("主题："+recordMetadata.topic());
        System.out.println("分区："+recordMetadata.partition());
        System.out.println("偏移量："+recordMetadata.offset());
        // 关闭生产者
        stringStringKafkaProducer.close();
    }
}
